package com.example.bigdata.spark.Actions算子

import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ListBuffer

object ActionsDemo {
    def main(args: Array[String]): Unit = {
        val sc = sparkContextInit()
        collectDemo(sc)   // 将RDD转换为Array数组
//        firstDemo(sc)     // 返回RDD 第一个元素
//        takeDemo(sc)      // 返回前num个元素并生成Array，它在获取元素时不会 对RDD中元素进行排序
//        topDemo(sc)   // 将RDD 中元素按照降序排列，然后返回前num 个元素， 可以在ord参数中指定其他排序规则
//        takeOrderedDemo(sc) // 默认情况下 top 方法按照K 进行降序排列，而不是根据V列进行排列
//        reduceDemo(sc)    // 聚合RDD集合中的每一个元素，聚合方式通过f函数来实现
//        aggregateDemo(sc) // 对元素进行聚合
//        foldDemo(sc) // 对元素进行聚合,与aggregate操作类似，是aggregate的简化版。
//        foreachDemo(sc)  // 遍历RDD中每个元素，应用f函数
//        foreachPartitionDemo(sc)
//        countDemo(sc)   // 计算RDD 中元素个数
//        lookupDemo(sc) // 查找 K值对应的所有V 值，并转换成Seq序列集合
//        countByKeyDemo(sc) // 统计每个K 值出现的次数
//        numRddDemo(sc)  // RDD数值操作
        saveAsTextFileDemo(sc) // RDD 存储
        sc.stop()

    }
    def sparkContextInit(): SparkContext ={
        val conf = new SparkConf()
                .setAppName("Actions 算子学习")
                .setMaster("local[1]")
        //                .setAppName("yarn")
//                .set("spark.submit.deployMode", "client")
        return new SparkContext(conf)
    }

    /**
      * def collect():Array[T]
      *     将RDD 转换为 Array数组
      */
    def collectDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(1 to 5)

        println(rdd1.collect().mkString(":"))
    }

    /**
      *  def first():T
      *     返回RDD 第一个元素
      */
    def firstDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(1 to 5)
        println(rdd1.first())
    }

    /**
      * def take(num:Int) : Array[T]
      *     返回前num个元素并生成Array，它在获取元素时不会 对RDD中元素进行排序
      */
    def takeDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(Array("Thomas","Alice","Kotlin"))
        println(rdd1.take(2).mkString(":"))
    }

    /**
      * def top(num:Int) (implicit ord:Ordering[T]):Array[T]
      *     将RDD 中元素按照降序排列，然后返回前num 个元素
      *     也可以在ord参数中指定其他排序规则
      *   举例: 返回考试分数前两名的学生
      */
    def topDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(Array(
            ("Alice",95),
            ("Tom",75),
            ("Thomas",88)
        ),2)
        // 默认情况下 top 方法按照K 进行降序排列，而不是根据V列进行排列
        println(rdd1.top(2)(Ordering.by(t=> t._2)).mkString(":"))
    }


    /**
      * def takeOrdered(num:Int) (implicit ord:Ordering[T]):Array[T]
      *     将RDD 中元素按照升序排列，然后返回前num 个元素
      *     也可以在ord参数中指定其他排序规则
      *   举例: 返回考试分数后两名的学生
      */
    def takeOrderedDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(Array(
            ("Alice",95),
            ("Tom",75),
            ("Thomas",88)
        ),2)
        // 默认情况下 top 方法按照K 进行降序排列，而不是根据V列进行排列
        println(rdd1.takeOrdered(2)(Ordering.by(t=> t._2)).mkString(":"))
    }

    /**
      * def reduce(f:(T,T) => T) :T
      *     聚合RDD集合中的每一个元素，聚合方式通过f函数来实现
      *  举例: 将每个科目和每个科目所对应的考试成绩进行聚合。
      */
    def reduceDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(Array(
            ("语文",95),
            ("数学",75),
            ("英语",88)
        ),2)
        // 结果 (String,Int)（语文_数学_英语,258）
        val rdd2 = rdd1.reduce((x,y) => {
            (x._1 +"_"+y._1,x._2+y._2)
        })
        println(rdd2.toString())
    }

    /**
      *  def aggregate[U:ClassTag](zeroValue:U)(seqOp:(U,T) => U,combOp:(U,U) =>U ):U
      *     对元素进行聚合
      *    zeroValue : 设置聚合的初始值。初始值的类型与RDD中元素类型不一致。
      *    seqOp : 将RDD 中每个元素聚合到类型为U的对象中。
      *    combOp : 跨分区聚合，对数据进行最终的汇总时调用此操作。
      *  举例: 将所有用户访问的URL聚合在一起(URL不去重)
      */
    def aggregateDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(Array(
        ("用户1","接口1"),
        ("用户2","接口1"),
        ("用户1","接口1"),
        ("用户1","接口2"),
        ("用户2","接口3")
    ),2)
        val rdd2 = rdd1.aggregate(ListBuffer[(String)]())(
            (list:ListBuffer[String],tuple:(String,String)) => list += tuple._2,
            (list1:ListBuffer[String],list2:ListBuffer[String]) => list1 ++= list2
        )
        rdd2.foreach(println(_))
    }

    /**
      * def fold(zeroValue:T)(op:(T,T) => T) :T
      *     对元素进行聚合,与aggregate操作类似，是aggregate的简化版。
      *     zeroValue:聚合时初始值，该值必须与RDD中元素类型一致
      *     op: 是一个函数，负责将RDD中元素聚合到zeroValue中，op函数用于将多个分区数据做最终聚合
      */
    def foldDemo(sc:SparkContext): Unit =
    {   // 2 * 1 + 1
        val rdd1 = sc.parallelize(Array(5,5,15,15) ,2)
        // x/y 代表多个分区，x+y 代表 分区数据求和
        val rdd2 = rdd1.fold(1)((x,y) => x + y)
        println(rdd2.toString)
    }

    /**
      *  def foreach(f:T => Unit) : Unit
      *    遍历RDD中每个元素,并依次应用f函数
      */
    def foreachDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(Array(
            ("用户1","接口1"),
            ("用户2","接口1"),
            ("用户1","接口1"),
            ("用户1","接口2"),
            ("用户2","接口3")
        ),2)
        rdd1.foreach(println)
    }

    /**
      *  def foreachPartition(t:Iterator[T] => Unit) : Unit
      *    该操作与foreach类似，但依次只遍历一个分区
      *    以分区为单位，遍历元素数据，foreach 以RDD 为单位遍历元素
      */
    def foreachPartitionDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(Array(5,5,15,15),2)
        rdd1.foreachPartition(iter =>{
            while (iter.hasNext){
                val element = iter.next()
                println(element+"---")
            }
        })
    }

    /**
      *  def count():Long
      *     计算RDD中的元素个数
      */
    def countDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(Array(
            ("语文",95),
            ("数学",75),
            ("英语",88)
        ),2)
        println("rdd1.count() : " + rdd1.count())
    }

    /**
      * def countByKey():Map[K,Long]
      *     统计RDD[K,V]集合中每个K出现的次数,最终返回Scala中的Map集合
      *
      *  举例: 统计每位用户调用过多少次接口
      */
    def countByKeyDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(Array(
            ("用户1","接口1"),
            ("用户2","接口1"),
            ("用户1","接口1"),
            ("用户1","接口2"),
            ("用户2","接口3")
        ),2)
        println("countByKey : "+rdd1.countByKey())
    }

    /**
      *  def lookup(key:K):Seq[V]
      *     指定RDD中元素的k值，返回RDD中k对应的所有元素的v值，并封装到Seq序列集合中
      */
    def lookupDemo(sc:SparkContext): Unit =
    {
        val rdd1 = sc.parallelize(Array(
            ("用户1","接口1"),
            ("用户2","接口1"),
            ("用户1","接口1"),
            ("用户1","接口2"),
            ("用户2","接口3")
        ),2)
        rdd1.cache()
        rdd1.unpersist()
        println(rdd1.lookup("用户1").mkString(":"))
    }

    /**
      *   spark 对数值类型的RDD 进行快速操作
      *     sum: 求和
      *     max: 最大值
      *     min:最小值
      *     mean: 平均值
      *     variance: 方差
      *     sampleVarance: 抽样方差
      *     stdev: 标准差
      *     sampleStdev: 抽样标准差
      */

    def numRddDemo(sc:SparkContext): Unit =
    {
        var numrdd1 = sc.parallelize(1 to 10)
        println("sum : "+numrdd1.sum())
        println("max : "+numrdd1.max())
        println("min : "+numrdd1.min())
        println("mean : "+numrdd1.mean())
        println("variance : "+numrdd1.variance())
        println("sampleVarance : "+numrdd1.sampleVariance())
        println("stdev : "+numrdd1.stdev())
        println("sampleStdev : "+numrdd1.sampleStdev())
    }

    def saveAsTextFileDemo(sc: SparkContext): Unit =
    {
        val rdd1 = sc.makeRDD(1 to 20)
        rdd1.saveAsTextFile("output/saveAsTextFileDemo")
        println("rdd1 : 保存完成 ！ ")
    }
}
